package com.zjs.jsp.status.messagelistener;

import com.currencyboot.service.rocketmq.messagelistener.Interface.MessageListenerConsumerInterface;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.zjs.jsp.domain.stats.JspStatus;
import com.zjs.jsp.domain.stats.OrderStatus;
import com.zjs.jsp.status.service.ProcessingService;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.List;
import java.util.UUID;

/**
 * @author 李文
 * @create 2018-07-18 14:13
 **/
public class ConsumableMessageListenerConsumer implements MessageListenerConsumerInterface
{
    private static final Logger LOGGER = LoggerFactory.getLogger(ConsumableMessageListenerConsumer.class);

    @Autowired
    ObjectMapper objectMapper;

    @Autowired
    ProcessingService pr;


    @Override
    public ConsumeConcurrentlyStatus consumeMessage(String s, MessageExt messageExt, ConsumeConcurrentlyContext consumeConcurrentlyContext) {

        long timestampt = LocalDateTime.now().atZone(ZoneId.systemDefault()).toInstant().toEpochMilli();
        //LOG.info("MQ生产信息时间 {},当前消费时间 {}", time, timestampt);
        String uuid = UUID.randomUUID().toString().replace("-", "");
        LOGGER.info("{} 接受的数据是 {} MQ生产信息时间 {},当前消费时间 {}  ",
                uuid, s, messageExt.getBornTimestamp(), timestampt);


        JspStatus mm;
        try {
            mm = objectMapper.readValue(s, JspStatus.class);
        } catch (Exception e) {
            LOGGER.error("{} 解析消息异常", uuid, e);
            return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
        }
        if (mm != null) {
            try {
                List<OrderStatus> jsp = pr.getOrderStatusList(uuid, mm, messageExt.getBornTimestamp());

                LOGGER.info("  {}  查询订单信息结果 {}",uuid,objectMapper.writeValueAsString(jsp));

                try {
                    // 成功失败返回处理判断
                    if (jsp != null && jsp.size() > 0) {
                        for (OrderStatus orderStatus : jsp) {
                            pr.sendOrderStatus(uuid, mm, orderStatus);
                        }
                    }
                } catch (org.springframework.dao.DuplicateKeyException e) {
                    LOGGER.warn("重复信息丢弃 " + mm.toString(), e);
                }

            } catch (Exception e) {
                LOGGER.error("{} 处理数据异常 ", uuid, e);
                return ConsumeConcurrentlyStatus.RECONSUME_LATER;
            }

        }
        return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
    }
}
